/**
 * Author:yuiop
 * Date:2022/10/20 15:18
 */
package com.devops.admin.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.devops.admin.config.bean.K8SClientInform;
import com.devops.admin.mapper.DOSK8sResourceEventMapper;
import com.devops.admin.model.*;
import com.devops.admin.po.DOSWebhookCustomInfoPo;
import com.devops.admin.redis.RedisService;
import com.devops.admin.service.*;
import com.devops.admin.util.*;
import com.devops.admin.vo.DOSNamespaceChart;
import com.devops.constant.DevOpsConstant;
import io.fabric8.kubernetes.api.model.*;
import io.fabric8.kubernetes.client.*;
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import io.sentry.connection.ConnectionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.SerializationUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.PipedOutputStream;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class DOSK8sResourceEventServiceImpl extends ServiceImpl<DOSK8sResourceEventMapper, DOSK8sResourceEvent> implements DOSK8sResourceEventService {
    @Autowired
    DOSReleaseOutService dosReleaseOutService;
    @Autowired
    DOSK8sPodService dosk8sPodService;
    @Autowired
    DOSProjectWebhookService dosProjectWebhookService;
    @Autowired
    K8SClientInform k8SClientInform;
    @Autowired
    DOSClusterInfoService dosClusterInfoService;

    @Autowired
    DOSWebhookCustomInfoService dosWebhookCustomInfoService;

    @Autowired
    DOSClusterService dosClusterService;

    @Autowired
    DOSClusterNamespaceService dosClusterNamespaceService;

    @Autowired
    DOSEnvClusterRelationService dosEnvClusterRelationService;

    @Autowired
    RedisService redisService;

    @Autowired
    DOSProjectOutService dosProjectOutService;

    @Value("${project.env}")
    String projectEnv;

    private final String REDIS_KEY_PRE = "DEV-OPS:WEBHOOK:";

    @Override
    public void watchAllClusterEvent() {
        List<DOSClusterInfo> clusterInfoList = dosClusterInfoService.getClusterInfoAndName();
        for (DOSClusterInfo clusterInfo : clusterInfoList) {
            KubernetesClient client = K8sClientUtil.getKubernetesClient(clusterInfo.getConfigPath(), clusterInfo.getConfigContent());
            try {
                client.v1().events().inform(dealResourceEventHandler(clusterInfo.getClusterId(), client));
            } catch (Exception e) {
                e.printStackTrace();
                if (e instanceof KubernetesClientException) {
                    log.error("********************集群不可用,clusterName=" + clusterInfo.getClusterName());
                }
            }
        }
    }

    @Override
    public void watchNamespaceEvent(Integer releaseId, Integer type, PipedOutputStream out) {
        //关闭此release的Inform 如果有的话
        k8SClientInform.closeAndRemoveInform(releaseId);
        DOSNamespaceChart namespaceChart = null;
        if (type == 1) {
            namespaceChart = dosReleaseOutService.getInfoByReleaseId(releaseId);
        } else if (type == 2) {
            namespaceChart = dosReleaseOutService.getInfoByResourceId(releaseId);
        }
        Assert.isTrue(namespaceChart != null, "请检查releaseId");
        KubernetesClient client = K8sClientUtil.getKubernetesClient(namespaceChart.getConfigPath(), namespaceChart.getConfigContent());
        SharedIndexInformer<Event> inform = client.v1().events().inNamespace(namespaceChart.getNamespace())
                .inform(dealNamespaceResourceEventHandler(namespaceChart.getClusterName(), out));
        k8SClientInform.addInform(releaseId, inform);
    }

    @Override
    public void closeInform(Integer releaseId) {
        // 关闭过长时间连接没有操作的
        k8SClientInform.closeAndRemoveInformByExpireTime();
        // 关闭指定的连接
        k8SClientInform.closeAndRemoveInform(releaseId);
    }

    private ResourceEventHandler<Event> dealNamespaceResourceEventHandler(String clusterName,
                                                                          PipedOutputStream out) {
        return new ResourceEventHandler<Event>() {
            @Override
            public void onAdd(Event obj) {
                getPodStatusByEvent(clusterName, obj, out);
            }

            @Override
            public void onUpdate(Event oldObj, Event newObj) {
                getPodStatusByEvent(clusterName, newObj, out);
            }

            @Override
            public void onDelete(Event obj, boolean deletedFinalStateUnknown) {

            }
        };
    }

    public void getPodStatusByEvent(String clusterName,
                                    Event resource,
                                    PipedOutputStream out) {
        try {
            String resourceName = resource.getInvolvedObject().getName();

            String resourceKind = resource.getInvolvedObject().getKind();
            // event
            String reason = resource.getReason();
            LocalDateTime eventTime = null;
            if (StrUtil.isNotBlank(resource.getLastTimestamp())) {
                eventTime = TimeUtil.utcStrToLocalDateTime(resource.getLastTimestamp());
            }
            String namespace = resource.getInvolvedObject().getNamespace();
            String eventType = resource.getType();
            String message = resource.getMessage();
            String msg = " \\x1B[32mTYPE:\\x1B[0m" + eventType + "  \\x1B[31mREASON:\\x1B[0m" + reason +
                    "  \\x1B[32m集群:\\x1B[0m" + clusterName +
                    "  \\x1B[31m命名空间:\\x1B[0m" + namespace +
                    "  \\x1B[32mKIND:\\x1B[0m" + resourceKind +
                    "  \\x1B[31mNAME:\\x1B[0m" + resourceName +
                    "  \\x1B[32mMessage:\\x1B[0m" + message +
                    "  \\x1B[31mEventTime:\\x1B[0m" + eventTime + System.lineSeparator();
            out.write(msg.getBytes());
            // 通知输出管道进行数据的读取
            out.flush();
        } catch (Exception e) {
            log.error("获取数据失败");
            e.printStackTrace();
        }
    }


    /**
     * 生成监听单个release event 的watcher
     * @return
     */
    public Watcher<Event> dealReleaseEventWatcher(Integer projectId, Integer clusterId, Integer releaseId, String releaseName, List<String> podNames) {
        return new Watcher<Event>() {

            @Override
            public void eventReceived(Action action, Event resource) {
                try {
                    String podName = resource.getInvolvedObject().getName();
                    if (podNames.contains(podName)) { //过滤pod事件
                        sendMsgAboutPod(resource, 1, clusterId, releaseId, releaseName, projectId);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onClose(WatcherException cause) {

            }
        };
    }

    /**
     * 生成监听整个集群event的handler 通知者模式
     *
     * @return
     */
    public ResourceEventHandler<Event> dealResourceEventHandler(Integer clusterId, KubernetesClient client) {
        return new ResourceEventHandler<Event>() {
            @Override
            public void onAdd(Event obj) {
                log.debug("集群{}新增event:{}", clusterId, JSONUtil.toJsonStr(obj));
                sendMsgPre(clusterId, client, obj);
            }

            @Override
            public void onUpdate(Event oldObj, Event newObj) {
                log.debug("集群{}更新event:old {}, new {}", clusterId, JSONUtil.toJsonStr(oldObj), JSONUtil.toJsonStr(newObj));
                sendMsgPre(clusterId, client, newObj);
            }

            @Override
            public void onDelete(Event obj, boolean deletedFinalStateUnknown) {

            }
        };
    }

    /**
     * 发送信息前置逻辑
     *
     */
    public void sendMsgPre(Integer clusterId, KubernetesClient client, Event resource) {
        try {
            String kind = resource.getInvolvedObject().getKind();
            String reason = resource.getReason();
            if ("pod".equalsIgnoreCase(kind)) {
                DOSRelease dosRelease = getProjectByEvent(resource, client);
                if (dosRelease != null && dosRelease.getId() > 0) {
                    sendMsgAboutPod(resource,
                            dosRelease.getReleaseType(),
                            clusterId,
                            dosRelease.getId(),
                            dosRelease.getReleaseName(),
                            dosRelease.getProjectId());
                } else {
                    sendMsgAboutPod(resource,
                            3,
                            clusterId,
                            null, null, null);
                }
                if ("started".equalsIgnoreCase(reason)) {
                    //reason还没见过为空的
                    if (StrUtil.isNotBlank(resource.getLastTimestamp())) {
                        LocalDateTime eventTime = TimeUtil.utcStrToLocalDateTime(resource.getLastTimestamp());
                        Duration duration = Duration.between(eventTime, LocalDateTime.now());
                        if (duration.getSeconds() < 30) {
                            CompletableFuture.runAsync(() -> {
                                sendMsgAboutPodRunning(client, clusterId, resource.getInvolvedObject().getNamespace(), resource.getInvolvedObject().getName());
                            });
                        }
                    }
                }

            } else {
                if (StrUtil.equalsAny(reason, "ResourcesSynced", "ResourceSyncAborted")
                        && (StrUtil.equalsAny(kind,"ApisixTls", "ApisixRoute", "ApisixIngress",
                        "ApisixUpstream", "ApisixPluginConfig"))) {
                    return;
                }
                sendMsgAboutOthers(resource, clusterId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 针对pod之外的其他类型产生的event 给管理员发送信息
     *
     */
    public void sendMsgAboutOthers(Event resource, Integer clusterId) {
        sendMsgAboutPod(resource, 4, clusterId, null, null, null);
    }

    /**
     * 针对pod event 给管理员、项目相关人员（如果有）发送信息
     *
     */
    public void sendMsgAboutPod(Event resource, Integer podType, Integer clusterId, Integer releaseId, String releaseName, Integer projectId) {
        String uid = resource.getMetadata().getUid();
        String resourceName = resource.getInvolvedObject().getName();
        String resourceKind = resource.getInvolvedObject().getKind();
        // event
        String reason = resource.getReason();
        String namespace = resource.getInvolvedObject().getNamespace();
        String eventType = resource.getType();
        String message = resource.getMessage();
        LocalDateTime eventTime = null;
        LambdaQueryWrapper<DOSK8sResourceEvent> wrapper = Wrappers.<DOSK8sResourceEvent>lambdaQuery().eq(DOSK8sResourceEvent::getUid, uid);
        if (StrUtil.isNotBlank(resource.getLastTimestamp())) {
            eventTime = TimeUtil.utcStrToLocalDateTime(resource.getLastTimestamp());
            wrapper.eq(DOSK8sResourceEvent::getEventTime, eventTime);
        }

        DOSK8sResourceEvent dbEvent = null;
        List<DOSK8sResourceEvent> dbEvents = baseMapper.selectList(wrapper);
        dbEvents = dbEvents.stream().sorted(Comparator.comparing(DOSK8sResourceEvent::getId).reversed()).collect(Collectors.toList());
        if (CollectionUtil.isNotEmpty(dbEvents)) {
            dbEvent = dbEvents.get(0);
            if (dbEvents.size() > 1) {
                log.error("********************数据库有重复event，uid=" + dbEvent.getUid());
            }

        }
        DOSCluster cluster = dosClusterService.getById(clusterId);
        if (dbEvent == null || dbEvent.getId() == null) {
            //数据库里没有9
            dbEvent = DOSK8sResourceEvent.builder()
                    .uid(uid)
                    .projectId(projectId)
                    .releaseId(releaseId)
                    .clusterName(cluster.getClusterName())
                    .namespace(namespace)
                    .podType(podType)
                    .resourceName(resourceName)
                    .resourceKind(resourceKind)
                    .reason(reason)
                    .eventTime(eventTime)
                    .message(message)
                    .eventType(eventType)
                    .eventContent(resource)
                    .pushStatus(1)
                    .build();
            baseMapper.insert(dbEvent);
        }
        if (dbEvent.getPushStatus() == 1) {
            //此resource event 未推送
            // 根据集群过滤webhook
            List<DOSWebhookCustomInfoPo> customInfoPos = dosProjectWebhookService.getWebhookAndCustom(projectId, clusterId);
            // 根据eventType过滤webhook
            List<DOSWebhookCustomInfoPo> customInfoPoList = filterWebhookByEventType(customInfoPos, eventType);
            if (customInfoPoList.size() > 0) {
                String title = eventType + "  " + (StrUtil.isNotBlank(releaseName) ? releaseName : resourceName) + "  " + resourceKind.toUpperCase() + "  " + reason;
                // 按照sendType的类型对webhook进行区分
                Map<Integer, List<DOSWebhookCustomInfoPo>> webhookCustomInfoMap = customInfoPoList.stream().collect(Collectors.groupingBy(DOSWebhookCustomInfoPo::getSendType));
                // 逐条发送
                if (CollectionUtil.isNotEmpty(webhookCustomInfoMap.get(2))) {
                    sendEventMsgByOne(resource, dbEvent, webhookCustomInfoMap.get(2));
                }
                // 合并发送
                if (CollectionUtil.isNotEmpty(webhookCustomInfoMap.get(1))) {
                    List<DOSWebhookCustomInfoPo> webhookCustomInfoPoList = new ArrayList<>();
                    for (DOSWebhookCustomInfoPo webhook : webhookCustomInfoMap.get(1)) {
                        webhookCustomInfoPoList.add(SerializationUtils.clone(webhook));
                    }
                    // 将事件缓存到redis
                    saveEventMessageToRedisAndSendEventInfo(SerializationUtils.clone(resource), SerializationUtils.clone(dbEvent),
                            webhookCustomInfoPoList);
                }
            }
        }

    }

    /**
     * @Author heyabo
     * @Description 将事件信息存入redis - 以Deployment作为hash类型的key，并且以事件产生的时间作为hash的key
     * @Date 11:26 2022/12/9
     * @Param [resource]
     **/
    private void saveEventMessageToRedisAndSendEventInfo(Event resource, DOSK8sResourceEvent dbEvent, List<DOSWebhookCustomInfoPo> webhooks) {
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String eventName = resource.getInvolvedObject().getName();
        String resourceKind = resource.getInvolvedObject().getKind();
        // 获取deployment名称
        String resourceName = getPodDeploy(eventName, resourceKind);
        // 根据event封装redis缓存信息
        JSONObject messageJson = getMessageJson(dbEvent);
        Map<Object, Object> redisValue = redisService.hGetAll(REDIS_KEY_PRE + resourceName.trim());
        // 暂存事件信息
        redisService.hSet(REDIS_KEY_PRE + resourceName.trim(), df.format(LocalDateTime.now()) , messageJson.toString());
        log.info(REDIS_KEY_PRE + resourceName.trim() + "保存到redis库中");
        if (CollectionUtil.isEmpty(redisValue)) {
            ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
            scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
                boolean bool = sendMsgFromQueryRedisByResourceName(resource, dbEvent, webhooks);
                if (bool) {
                    scheduledThreadPoolExecutor.shutdown();
                }
            }, 6, 6, TimeUnit.SECONDS);
        }

    }

    /**
     * @return void
     * @Author heyabo
     * @Description 通过redis查询事件的信息并发送
     * @Date 15:15 2022/12/9
     * @Param [resource]
     **/
    private boolean sendMsgFromQueryRedisByResourceName(Event resource, DOSK8sResourceEvent dbEvent, List<DOSWebhookCustomInfoPo> webhooks) {
        boolean bool = false;
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String eventName = resource.getInvolvedObject().getName();
        String resourceKind = resource.getInvolvedObject().getKind();
        String resourceName = getPodDeploy(eventName, resourceKind);
        String msgTitle = getMsgTitle(resourceName, dbEvent);
        String msgTemplate;
        String lastTime = "";
        for (DOSWebhookCustomInfoPo webhook : webhooks) {
            try {
                long seconds;
                log.info("开始webhook发送前的准备工作...");
                Map<String, Object> result = getEventInfoFromRedis(resourceName);
                if (CollectionUtil.isEmpty(result)) {
                    redisService.del(REDIS_KEY_PRE + resourceName.trim());
                    log.info(REDIS_KEY_PRE + resourceName.trim() + "已经被删除");
                    return true;
                }
                // 根据名称将集合转换成map方便接下来的判断操作
                Map<String, String> resultMap = setMessageTemplateInfo(result, webhook, msgTitle);
                if (CollectionUtil.isEmpty(resultMap)) {
                    // 如果当前的模板信息为空则没有必要再往下执行了，直接下一个webhook走起
                    break;
                }
                msgTemplate = resultMap.get("msgTemplate");
                lastTime = resultMap.get("lastTime");
                log.info("获取到发送webhook的消息模板");

                LocalDateTime now = LocalDateTime.now();
                // 如果最后一条保存到redis的事件的保存事件与当前的事件相差3秒以上，则发送消息
                seconds = Duration.between(LocalDateTime.parse(lastTime, df), now).getSeconds();
                log.info(String.valueOf(seconds));
                if (seconds >= 5) {
                    // 注释掉此块代码 因程序执行时间可忽略不计 若在程序执行期间真产生新数据 做丢弃处理
                    // 多线程的情况下判断当前的key是否还在redis中，如果已经删除则没有必要往下走了，直接返回结束线程池
                    // 再次获取排序后的事件消息集合
//                    Map<String, Object> objectObjectMap = getEventInfoFromRedis(resourceName);
//                    if (CollectionUtil.isNotEmpty(objectObjectMap) && result.size() > 0) {
//                        // 如果最新获取的数据比以前的多则需要使用最新获取的数据
//                        if (objectObjectMap.size() > result.size()) {
//                            Map<String, String> resultNewMap = setMessageTemplateInfo(objectObjectMap, webhook, msgTitle);
//                            msgTemplate = resultNewMap.get("msgTemplate");
//                            lastTime = resultNewMap.get("lastTime");
//                            seconds = Duration.between(LocalDateTime.parse(lastTime, df), now).getSeconds();
//                            if (seconds < 5) {
//                                return false;
//                            }
//                        }
//                    }
                    String webhookUrl = WebhookUtil.dealWebHook(webhook.getWebhook(), webhook.getSecret());
                    sendWebhookMessage(webhookUrl, msgTemplate);
                    bool = true;
                    log.info(webhook.getTitle() + "-消息发送成功：" + msgTemplate);
                }
            } catch (Exception e) {
                redisService.del(REDIS_KEY_PRE + resourceName.trim());
                log.info(REDIS_KEY_PRE + resourceName.trim() + "已经被删除");
            }
        }
        if (bool) {
            redisService.del(REDIS_KEY_PRE + resourceName.trim());
            log.info(REDIS_KEY_PRE + resourceName.trim() + "已经被删除");
            dbEvent.setPushStatus(2);
            baseMapper.updateById(dbEvent);
        }
        return bool;
    }

    /**
     * @return java.util.Map<java.lang.String, java.lang.String>
     * @Author heyabo
     * @Description 为模板设置信息
     * @Date 15:23 2022/12/12
     * @Param [result]
     **/
    private Map<String, String> setMessageTemplateInfo(Map<String, Object> result,
                                                       DOSWebhookCustomInfoPo customInfoByWebhook, String msgTitle) {

        Map<String, String> resultMap = new HashMap<>();
        Map<String, Object> sendMap = new HashMap<>();
        // 事件开始的真实时间
        String startTime = "";
        // 事件结束的真实时间
        String endTime = "";
        // 最后一条数据到redis的时间
        String lastTime = "";
        StringBuilder message = new StringBuilder();
        // 拼接发送信息
        for (Map.Entry<String, Object> entry : result.entrySet()) {
            lastTime = entry.getKey();
            JSONObject json = JSONUtil.parseObj(entry.getValue());
            sendMap.put(json.get("eventTime").toString(), entry.getValue());
        }
        // 根据事件发生的先后顺序进行排序
        LinkedHashMap<String, Object> map = sendMap.entrySet().stream().sorted(Map.Entry.comparingByKey())
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue,
                        (oldValue, newValue) -> oldValue, LinkedHashMap::new));
        // 当前webhook存在模板
        if (StrUtil.isNotEmpty(customInfoByWebhook.getMsgTemplate())) {
            // 获取webhook的配置信息中模板
            String webhookMsgTemplate = customInfoByWebhook.getMsgTemplate();
            // 将模板转换成json
            JSONObject jsonObject = JSONUtil.parseObj(webhookMsgTemplate);
            String text = webhookMsgTemplate;
            if (jsonObject.containsKey("markdown")) {
                String markdown = jsonObject.get("markdown").toString();
                jsonObject.toString();
                JSONObject markdownJson = JSONUtil.parseObj(markdown);
                Assert.isTrue(markdownJson.containsKey("text"), "json格式不正确，无法解析...");
                text = markdownJson.get("text").toString();
            }
            // 循环从redis中查询到的事件信息，并根据webhook的配置信息拼接模版消息
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                JSONObject json = JSONUtil.parseObj(entry.getValue());
                if (StrUtil.isEmpty(startTime)) {
                    startTime = json.get("eventTime").toString();
                }
                endTime = json.get("eventTime").toString();
                String msg = setCustomInfoMsg(json, text);
                message.append("> - ").append(msg).append(System.lineSeparator());
            }
        } else {
            // 不存webhook的配置信息
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                JSONObject json = JSONUtil.parseObj(entry.getValue());
                if (StrUtil.isEmpty(startTime)) {
                    startTime = json.get("eventTime").toString();
                }
                endTime = json.get("eventTime").toString();
                String text = "**TYPE**:#{type}  **REASON**:#{reason}  **集群**:#{cluster}  " +
                        "**命名空间**:#{namespace}  **KIND**:#{kind}  **NAME**:#{name}  **Message**:#{message}  **EventTime**:#{eventTime}";
                String msg = setCustomInfoMsg(json, text);
                message.append("> - ").append(msg).append(System.lineSeparator());
            }
        }

        String singleLineStr = "##### " + msgTitle + "\n"
                + "> - " + startTime + " \n"
                + message + "> - " + endTime;
        JSONObject msgJson = new JSONObject();
        //md格式
        JSONObject mdJson = new JSONObject();
        mdJson.set("title", msgTitle);
        mdJson.set("text", singleLineStr);
        msgJson.set("msgtype", "markdown");
        msgJson.set("markdown", mdJson);

        resultMap.put("msgTemplate", msgJson.toString());
        resultMap.put("lastTime", lastTime);

        return resultMap;
    }

    /**
     * @return java.lang.String
     * @Author heyabo
     * @Description 组装自定义模板的信息
     * @Date 16:11 2022/12/12
     * @Param [json, text]
     **/
    private String setCustomInfoMsg(JSONObject json, String text) {

        String resultStr = text;
        if (text.contains("#{type}")) {
            resultStr = resultStr.replace("#{type}", json.getStr("type", ""));
        }
        if (text.contains("#{reason}")) {
            resultStr = resultStr.replace("#{reason}", json.getStr("reason", ""));
        }
        if (text.contains("#{cluster}")) {
            resultStr = resultStr.replace("#{cluster}", json.getStr("cluster", ""));
        }
        if (text.contains("#{namespace}")) {
            resultStr = resultStr.replace("#{namespace}", json.getStr("namespace", ""));
        }
        if (text.contains("#{kind}")) {
            resultStr = resultStr.replace("#{kind}", json.getStr("kind", ""));
        }
        if (text.contains("#{name}")) {
            resultStr = resultStr.replace("#{name}", json.getStr("name", ""));
        }
        if (text.contains("#{message}")) {
            resultStr = resultStr.replace("#{message}", json.getStr("message", ""));
        }
        if (text.contains("#{eventTime}")) {
            resultStr = resultStr.replace("#{eventTime}", json.getStr("eventTime", ""));
        }

        return resultStr;
    }

    /**
     * @return java.util.Map<java.lang.String, java.lang.Object>
     * @Author heyabo
     * @Description 从redis中获取暂存的事件消息集合并按照key正序返回
     * @Date 09:55 2022/12/12
     * @Param [resourceName]
     **/
    private Map<String, Object> getEventInfoFromRedis(String resourceName) {
        Map<Object, Object> msgMap = redisService.hGetAll(REDIS_KEY_PRE + resourceName.trim());
        if (CollectionUtil.isEmpty(msgMap)) {
            return null;
        }
        Map<String, Object> map = new HashMap<>();
        msgMap.forEach((k, v) -> map.put(k.toString(), v));
        // 对查询的结果进行排序
        Map<String, Object> result = map.entrySet().stream().sorted(Map.Entry.comparingByKey())
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue,
                        (oldValue, newValue) -> oldValue, LinkedHashMap::new));
        return result;
    }

    /**
     * @return java.lang.String
     * @Author heyabo
     * @Description 拼接消息的标题
     * @Date 09:39 2022/12/12
     * @Param [resourceName, dbEvent]
     **/
    private String getMsgTitle(String resourceName, DOSK8sResourceEvent dbEvent) {
        DOSProject project = dosProjectOutService.getById(dbEvent.getProjectId());
        String projectName = Optional.ofNullable(dbEvent.getNamespace()).orElse("");
        if (project != null && project.getId() > 0) {
            projectName = project.getProjectName();
        }
        DOSRelease release = dosReleaseOutService.getReleseById(dbEvent.getReleaseId());
        String releaseName = getPodDeploy(resourceName, dbEvent.getResourceKind());
        if (release != null && release.getId() > 0) {
            releaseName = release.getReleaseName();
        }
        // 如果项目名称和releaseName都为空
        /*if (StrUtil.isEmpty(projectName) || StrUtil.isEmpty(releaseName)) {
            // 先截取resource的名称得到releaseName
            releaseName = splicingName(resourceName, 1);
            release = dosReleaseOutService.getReleaseByReleaseName(releaseName);
            // 查询到release则赋值
            if (release != null && release.getProjectId() > 0) {
                projectName = getProjectName(release.getProjectId());
            } else {
                // 不对resourceName进行截取，直接使用resourceName到数据库中查询release
                release = dosReleaseOutService.getReleaseByReleaseName(resourceName);
                releaseName = resourceName;
                // 查询到release则对项目名称赋值
                if (release != null && release.getProjectId() > 0) {
                    projectName = getProjectName(release.getProjectId());
                } else {
                    // 查询不到，则到yml文件部署的表里查询
                    DOSKubernetesResource kubernetesResource = dosReleaseOutService.getKubernetesResourceByResourceName(resourceName);
                    if (kubernetesResource != null && kubernetesResource.getId() > 0) {
                        projectName = getProjectName(kubernetesResource.getProjectId());
                    }
                }
            }
        }*/
        return dbEvent.getClusterName() + "  " + projectName + "  " + releaseName;
    }

    private String getProjectName(Integer projectId) {

        String projectName = "";
        DOSProject project = dosProjectOutService.getById(projectId);
        if (project != null && project.getId() > 0) {
            projectName = project.getProjectName();
        }
        return projectName;
    }

    @NotNull
    private static JSONObject getMessageJson(DOSK8sResourceEvent dbEvent) {
        String eventTime;
        JSONObject messageJson = new JSONObject();
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        messageJson.set("type", dbEvent.getEventType());
        messageJson.set("reason", dbEvent.getReason());
        messageJson.set("cluster", dbEvent.getClusterName());
        messageJson.set("namespace", dbEvent.getNamespace());
        messageJson.set("kind", dbEvent.getResourceKind());
        messageJson.set("name", dbEvent.getResourceName());
        messageJson.set("message", dbEvent.getMessage());
        if (dbEvent.getEventTime() != null) {
            eventTime = df.format(dbEvent.getEventTime());
        } else {
            eventTime = df.format(LocalDateTime.now());
        }
        messageJson.set("eventTime", eventTime);
        return messageJson;
    }

    /**
     * @return void
     * @Author heyabo
     * @Description 根据kind的类型获取deployment名称
     * @Date 11:30 2022/12/9
     * @Param [eventName, resourceKind]
     **/
    private String getPodDeploy(String eventName, String kind) {
        String resourceName = eventName;
        if ("Pod".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 2);
        } else if ("ReplicaSet".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 1);
        } else if ("Deployment".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 0);
        } else if ("CertificateRequest".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 1);
        } else if ("Order".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 2);
        } else if ("Challenge".equalsIgnoreCase(kind)) {
            resourceName = splicingName(eventName, 3);
        }
        return resourceName;
    }

    private static String splicingName(String eventName, int num) {
        String resourceName;
        String[] split = eventName.split("-");
        int newArrayLength = Math.max(split.length - num, 1);
        resourceName = StrUtil.join("-", Arrays.copyOf(split, newArrayLength));
        return resourceName;
    }

    public void sendMsgAboutPodRunning(KubernetesClient client, Integer clusterId, String namespace, String podName) {
        boolean skipFlag = false;
        int recycleCount = 1;
        while (!skipFlag && recycleCount < 20) {
            //如果一分钟内都转不到running状态，大概率出问题了，结束循环
            recycleCount++;
            try {
                Pod pod = client.pods().inNamespace(namespace).withName(podName).get();
                if (pod == null) {
                    //一次性任务的pod，可能运行完直接删除了
                    return;
                }
                PodStatus podStatus = pod.getStatus();
                List<ContainerStatus> containerStatus = podStatus.getContainerStatuses();
                if (CollectionUtil.isNotEmpty(containerStatus)) {
                    ContainerStatus firstContainerStatus = containerStatus.get(0);
                    if (firstContainerStatus != null && firstContainerStatus.getReady()) {
                        skipFlag = true;
                        LambdaQueryWrapper<DOSProjectWebhook> queryWrapper;
                        queryWrapper = Wrappers.<DOSProjectWebhook>lambdaQuery().eq(DOSProjectWebhook::getType, 1)
                                .eq(DOSProjectWebhook::getSubType, 3)
                                .eq(DOSProjectWebhook::getStatus, 1);
                        List<DOSProjectWebhook> webhooks = dosProjectWebhookService.list(queryWrapper);
                        DOSCluster cluster = dosClusterService.getById(clusterId);
                        for (DOSProjectWebhook webhook : webhooks) {
                            String webhookUrl = WebhookUtil.dealWebHook(webhook.getWebhook(), webhook.getSecret());
                            JSONObject msgJson = new JSONObject();
                            msgJson.set("msgtype", "text");
                            JSONObject textJson = new JSONObject();
                            textJson.set("content", cluster.getClusterName() + " Pod " + podName + " 已running");
                            msgJson.set("text", textJson);
                            if (sendWebhookMessage(webhookUrl, msgJson.toString())) {
                                log.info("****************发送成功");
                            }
                        }
                    }
                }
                Thread.sleep(3000L);
            } catch (Exception e) {
                //一旦发生错误 大概率是pod不存在了 结束循环
                e.printStackTrace();
                skipFlag = true;
            }

        }
    }

    /**
     * 根据event获取触发事件的项目
     *
     * @param event
     * @param client
     * @return
     */
    public DOSRelease getProjectByEvent(Event event, KubernetesClient client) {

        DOSRelease dosRelease = null;
        // 获取event的kind，如果是pod进行处理，不是pod待定
        // 最后获取的是projectId
        String kind = event.getInvolvedObject().getKind();
        String aCase = kind.toLowerCase();
        if (aCase.equals("pod")) {
            String podName = event.getInvolvedObject().getName();
            String namespace = event.getInvolvedObject().getNamespace();
            // 根据podName获取pod实列
            Pod pod = client.pods().inNamespace(namespace).withName(podName).get();
            if (pod != null) {
                Map<String, String> labels = pod.getMetadata().getLabels();
                // helm方式部署的项目
                if (labels.containsKey(DevOpsConstant.K8sConfig.POD_LABEL_RELEASE_NAME)) {
                    // 根据labels的key获取release名称
                    String releaseName = labels.get(DevOpsConstant.K8sConfig.POD_LABEL_RELEASE_NAME);
                    if (StrUtil.isNotBlank(releaseName)) {
                        if (labels.containsKey(DevOpsConstant.K8sConfig.K8S_LABEL_APPLY_TYPE)) {
                            // yaml方式部署
                            dosRelease = getDosRelease(Wrappers.<DOSKubernetesResource>lambdaQuery()
                                    .eq(DOSKubernetesResource::getName, releaseName));
                        } else {
                            // 根据releaseName获取helm方式部署的release的详细信息
                            List<DOSRelease> releases = dosReleaseOutService.selectListByReleaseName(releaseName);
                            if (CollectionUtil.isNotEmpty(releases)) {
                                if (releases.size() > 1) {
                                    return null;
                                }
                                dosRelease = releases.get(0);
                                dosRelease.setReleaseType(1);
                            }
                        }
                    }

                } else {
                    // yaml方式部署的项目 - 兼容以前没有设置labels的可以的数据
                    List<OwnerReference> ownerReferences = pod.getMetadata().getOwnerReferences();
                    if (CollectionUtil.isNotEmpty(ownerReferences)) {
                        String name = ownerReferences.get(0).getName();
                        if (labels.containsKey(DevOpsConstant.K8sConfig.POD_TEMPLATE_HASH)) {
                            String podTemplateHash = labels.get(DevOpsConstant.K8sConfig.POD_TEMPLATE_HASH);
                            String kubernetesResourceName = name.replace("-" + podTemplateHash, "").trim();
                            dosRelease = getDosRelease(Wrappers.<DOSKubernetesResource>lambdaQuery()
                                    .eq(DOSKubernetesResource::getName, kubernetesResourceName));
                        }
                    }
                }

            }
        }

        return dosRelease;
    }

    private DOSRelease getDosRelease(LambdaQueryWrapper<DOSKubernetesResource> kubernetesResourceName) {
        DOSRelease dosRelease = null;
        List<DOSKubernetesResource> kubernetesResources = dosReleaseOutService.getK8sList(kubernetesResourceName);
        if (CollectionUtil.isNotEmpty(kubernetesResources)) {
            if (kubernetesResources.size() > 1) {
                return null;
            }
            DOSKubernetesResource kubernetesResource = kubernetesResources.get(0);
            dosRelease = DOSRelease.builder()
                    .id(kubernetesResource.getId())
                    .releaseName(kubernetesResource.getName())
                    .projectId(kubernetesResource.getProjectId())
                    .releaseType(2)
                    .clusterId(kubernetesResource.getClusterId())
                    .namespaceId(kubernetesResource.getNamespaceId())
                    .build();
        }
        return dosRelease;
    }

    private boolean sendWebhookMessage(String url, String message) {
        HttpResponse response = HttpRequest.post(url).body(message
                , "application/json;charset=UTF-8")
                .timeout(6000)
                .execute();
        if (response != null) {
            return response.getStatus() == 200;
        }

        return false;
    }

    /**
     * @return java.util.List<com.devops.admin.po.DOSWebhookCustomInfoPo>
     * @Author heyabo
     * @Description 根据eventType过滤webhook
     * @Date 14:41 2023/1/9
     * @Param [customInfoPos, eventType]
     **/
    private List<DOSWebhookCustomInfoPo> filterWebhookByEventType(List<DOSWebhookCustomInfoPo> customInfoPos, String eventType) {
        List<DOSWebhookCustomInfoPo> customInfoPoList = new ArrayList<>();
        // 首先判断一下webhook的集合是否为空
        if (CollectionUtil.isNotEmpty(customInfoPos)) {
            for (DOSWebhookCustomInfoPo customInfoPo : customInfoPos) {
                // 首先设置一下发送类型，发送类型为空的都是默认的发送方式 -> 合并发送
                if (customInfoPo.getSendType() == null || customInfoPo.getSendType() < 1) {
                    customInfoPo.setSendType(1);
                }
                // 如果过滤类型不为空则判断当前事件类型是否符合发送要求
                if (StrUtil.isNotEmpty(customInfoPo.getFilterType())) {
                    if (customInfoPo.getFilterType().contains(eventType.toLowerCase())) {
                        customInfoPoList.add(customInfoPo);
                    }
                } else {
                    // 如果为空则一定符合条件不需要再比对了
                    customInfoPoList.add(customInfoPo);
                }
            }
        }

        return customInfoPoList;
    }

    /**
     * @return void
     * @Author heyabo
     * @Description 逐条发送事件信息
     * @Date 14:57 2023/1/9
     * @Param [resource, dbEvent, customInfoPos, title]
     **/
    private void sendEventMsgByOne(Event resource, DOSK8sResourceEvent dbEvent, List<DOSWebhookCustomInfoPo> customInfoPos) {
        JSONObject messageJson = getMessageJson(dbEvent);
        String eventName = resource.getInvolvedObject().getName();
        String resourceKind = resource.getInvolvedObject().getKind();
        String resourceName = getPodDeploy(eventName, resourceKind);
        String msgTitle = getMsgTitle(resourceName, dbEvent);
        boolean bool = false;
        // 循环webhook发送消息
        for (DOSWebhookCustomInfoPo customInfoPo : customInfoPos) {
            try {
                String msg;
                if (StrUtil.isNotEmpty(customInfoPo.getMsgTemplate())) {
                    // 获取webhook的配置信息中模板
                    String webhookMsgTemplate = customInfoPo.getMsgTemplate();
                    // 将模板转换成json
                    JSONObject jsonObject = JSONUtil.parseObj(webhookMsgTemplate);
                    if (jsonObject.containsKey("markdown")){
                        String markdown = jsonObject.get("markdown").toString();
                        JSONObject markdownJson = JSONUtil.parseObj(markdown);
                        Assert.isTrue(markdownJson.containsKey("text"),"json格式不正确，无法解析...");
                        String text = markdownJson.get("text").toString();
                        msg = "> - " + setCustomInfoMsg(messageJson, text);
                    } else {
                        // 非默认的json格式，直接进行replace替换
                        String eventMsg = setCustomInfoMsg(messageJson, webhookMsgTemplate);
                        if (eventMsg.contains("#{title}")){
                            eventMsg = eventMsg.replace("#{title}",msgTitle);
                        }
                        msg = eventMsg;
                    }

                } else {
                    // 如果不存在配置模板使用默认的模板，就是把所有的信息都发送
                    String text = "**TYPE**:#{type}  **REASON**:#{reason}  **集群**:#{cluster}  " +
                            "**命名空间**:#{namespace}  **KIND**:#{kind}  **NAME**:#{name}  **Message**:#{message}  **EventTime**:#{eventTime}";
                    msg = "> - " + setCustomInfoMsg(messageJson, text);

                }

                bool = buildMsgAndSend(messageJson, msgTitle, customInfoPo, msg);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (bool) {
            dbEvent.setPushStatus(2);
            baseMapper.updateById(dbEvent);
        }
    }

    private boolean buildMsgAndSend(JSONObject messageJson, String msgTitle, DOSWebhookCustomInfoPo customInfoPo, String msg) throws Exception {
        boolean bool = false;
        try{
            String singleLineStr = msg;
            if (msg.contains(">")) {
                singleLineStr = "##### " + msgTitle + "\n"
                        + "> - " + messageJson.get("eventTime").toString() + "\n"
                        + msg;
            }
            JSONObject msgJson = new JSONObject();
            //md格式
            JSONObject mdJson = new JSONObject();
            mdJson.set("title", msgTitle);
            mdJson.set("text", singleLineStr);
            msgJson.set("msgtype", "markdown");
            msgJson.set("markdown", mdJson);
            String webhookUrl = WebhookUtil.dealWebHook(customInfoPo.getWebhook(), customInfoPo.getSecret());
            bool = sendWebhookMessage(webhookUrl, msgJson.toString());
            log.info("逐条消息发送成功: " + msgJson);
        }catch (Exception e){
            log.error("webhook事件消息发送失败...");
            e.printStackTrace();
        }

        return bool;
    }

}

